import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.math3.fitting.leastsquares.EvaluationRmsChecker;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.mortbay.util.ajax.JSON;

import java.math.BigDecimal;
import java.util.Properties;

public class FlinkKafkaJson {
    public static void main(String[] args) throws Exception {

        //设置消费者
        String topic1 = "druid1";
        Properties props1 = new Properties();
        props1.setProperty("bootstrap.servers","centos7-1:9092");
        props1.setProperty("group.id","mygp");
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<String>(topic1, new SimpleStringSchema(), props1);
        consumer.setStartFromGroupOffsets();
        //设置生产者
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<String>("centos7-1:9092","druid2",new SimpleStringSchema());


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> data = env.addSource(consumer);

        SingleOutputStreamOperator<String> outStream = data.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                //提取json中非数组字段
                Long ts = (Long) jsonObject.get("ts");
                String orderId = (String) jsonObject.get("orderId");
                String userId = (String) jsonObject.get("userId");
                Integer orderStatusId = (Integer) jsonObject.get("orderStatusId");
                String orderStatus = (String) jsonObject.get("orderStatus");
                Integer payModeId = (Integer) jsonObject.get("payModeId");
                String payMode = (String) jsonObject.get("payMode");
                String payment = (String) jsonObject.get("payment");
                //提取json中数组
                JSONArray products = jsonObject.getJSONArray("products");
                for (Object product : products) {
                    JSONObject outJsonObject = new JSONObject();
                    outJsonObject.put("ts", ts);
                    outJsonObject.put("orderId", orderId);
                    outJsonObject.put("userId", userId);
                    outJsonObject.put("orderStatusId", orderStatusId);
                    outJsonObject.put("orderStatus", orderStatus);
                    outJsonObject.put("payModeId", payModeId);
                    outJsonObject.put("payMode", payMode);
                    outJsonObject.put("payment", payment);

                    JSONObject productJsonObject = (JSONObject) product;
                    String productId = (String) productJsonObject.get("productId");
                    outJsonObject.put("productId", productId);
                    String productName = (String) productJsonObject.get("productName");
                    outJsonObject.put("productName", productName);
                    BigDecimal price = (BigDecimal) productJsonObject.get("price");
                    outJsonObject.put("price", price);
                    Integer productNum = (Integer) productJsonObject.get("productNum");
                    outJsonObject.put("productNum", productNum);
                    String categoryid = (String) productJsonObject.get("categoryid");
                    outJsonObject.put("categoryid", categoryid);
                    String catname1 = (String) productJsonObject.get("catname1");
                    outJsonObject.put("catname1", catname1);
                    String catname2 = (String) productJsonObject.get("catname2");
                    outJsonObject.put("catname2", catname2);
                    String catname3 = (String) productJsonObject.get("catname3");
                    outJsonObject.put("catname3", catname3);
                    String outStr = outJsonObject.toString();
                    out.collect(outStr);
                }
            }
        });
        outStream.addSink(producer);
        env.execute();
    }
}
